package cn.ichiva.dl.tushare.analy.ws;

import cn.ichiva.common.DataHelper;
import cn.ichiva.common.io.IOHelper;
import cn.ichiva.dl.tushare.analy.impl.TradeAgent;
import cn.ichiva.dl.tushare.analy.impl.TradeAgentList;
import cn.ichiva.dl.tushare.analy.TrainHelper;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.io.Serializable;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;


/**
 * Created by game group on 2020/12/8.
 */
@Slf4j
public class AgentTrainClient {

    public static void main(String[] args) {
        String wsUri = "ws://127.0.0.1:10948";
        if(null != args && args.length > 0){
            wsUri = args[0];
        }

        WebSocketTrainClient ws =  new WebSocketTrainClient(URI.create(wsUri));
        ws.connect();

        while (!Thread.currentThread().isInterrupted()){
            try {
                TradeAgent agent = ws.queue.poll();
                if(null == agent){
                    log.info("空闲");
                    Thread.sleep(1000);
                    continue;
                }

                ws.train(agent);
            }catch (InterruptedException e){
                Thread.currentThread().interrupt();
            }catch (Exception ex){
                log.error(ex.getMessage(),ex);
            }
        }
    }

    private static  class WebSocketTrainClient extends WebSocketClient {
        private TradeAgentList agentList;
        public final ArrayBlockingQueue<TradeAgent> queue = new ArrayBlockingQueue<>(2);

        public WebSocketTrainClient(URI serverUri) {
            super(serverUri);
        }

        @Override
        public void onOpen(ServerHandshake handshakedata) {
            log.info("已连接...");

            //初始化排行榜
            WSRequest req = new WSRequest();
            req.setAction(ApiEnum.AGENT_LIST.getAction());
            sendObject(req);

            new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()){
                    try {
                        if(null == agentList){
                            log.info("未初始化");
                            Thread.sleep(1000);
                            continue;
                        }

                        queue.put(agentList.getNewAgent());
                    }catch (InterruptedException e){
                        Thread.currentThread().interrupt();
                    }catch (Exception ex){
                        log.error(ex.getMessage(),ex);
                    }
                }
            }).start();
        }

        private void sendObject(Object req) {
            send(DataHelper.toByte((Serializable) req));
        }

        public void train(TradeAgent agent) {
            if (TrainHelper.contend(agent)) {
                WSRequest req = new WSRequest();
                req.setAction(ApiEnum.TRAIN_RES.getAction());
                req.setData(agent);
                send(DataHelper.toByte(req));

                log.info("第三阶段 agent = {}",agent);
            }
        }

        @Override
        public void onMessage(ByteBuffer bytes) {
            byte[] array = bytes.array();
            WSResponse res = (WSResponse) DataHelper.toObject(array);

            String action = res.getAction();
            if(ApiEnum.AGENT_LIST.getAction().equals(action)){
                Collection<TradeAgent> list = (List<TradeAgent>) res.getData();
                this.agentList = new TradeAgentList(list);
            }else if(ApiEnum.PUT_LIST.getAction().equals(action)){
                List<TradeAgent> list = (List<TradeAgent>) res.getData();
                list.forEach(this.agentList::put);
            }
        }

        @Override
        public void onMessage(String message) {
            log.info("message = {}",message);
        }

        @Override
        public void onClose(int code, String reason, boolean remote) {
            log.info("断开!");
        }

        @Override
        public void onError(Exception ex) {
            log.error(ex.getMessage(),ex);
        }
    }

}
